package com.itunion.quickstart.stream;

import com.itunion.quickstart.pojo.Goods;
import com.itunion.quickstart.pojo.Student;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *
 * 功能描述: 编写第一个HelloWorld，实时计算商品点击数量
 *
 * @auther: lin
 * @date: 2021/9/23 10:32 上午
 */
public class HelloWorldGoods {
    public static void main(String[] args) throws Exception {
        // TODO 第一步：创建流计算执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // TODO 第二步：实时接收输入数据
        DataStream<String> inputDataStream = env.socketTextStream("127.0.0.1",7777);
        // TODO 第三步：转换数据类型
        DataStream<Goods> dataStream = inputDataStream.map(new MapFunction<String, Goods>() {
            @Override
            public Goods map(String s) throws Exception {
                String[] fields = s.split(",");
                return new Goods(fields[0],Integer.parseInt(fields[1]));
            }
        });
        // TODO 第四步：根据商品名称进行分组
        KeyedStream<Goods,String> keyedStream = dataStream.keyBy(goods -> goods.getGoodsName());
        // TODO 第五步：滚动聚合，计算每个商品的总点击量
        DataStream<Goods> resultStream = keyedStream.sum("clickNum");
        // TODO 第六步：打印输出
        resultStream.print();
        // TODO 第七步：执行
        env.execute();
    }
}
